package com.zdb.demo.flink.apitest

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011}
import org.apache.flink.streaming.api.scala._

object Sensor {

  // 定义一个数据样例类，传感器ID,采集时间戳，传感器温度
  case class SensorReading(id: String, timestamp: Long, temperature: Double)

  def main(args: Array[String]): Unit = {
    val conf:Configuration = new Configuration()
      conf.setInteger("rest.port", 9191)
    conf.setBoolean("local.start-webserver", true)

    // 1. 获取执行环境
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf)
//    env.setParallelism(10)
    env.enableCheckpointing(5000)

    // 2. Source
    // Source1: 从集合读取

//    val stream1:DataStream[SensorReading] = env.fromCollection[SensorReading](List[SensorReading](
//      SensorReading("sensor_1", 1547718199, 35.80018327300259),
//        SensorReading("sensor_6", 1547718201, 15.402984393403084),
//        SensorReading("sensor_7", 1547718202, 6.720945201171228),
//        SensorReading("sensor_10", 1547718205, 38.101067604893444)
//    ))

    // Source2: 从文件读取
//    val steam2 = env.readTextFile("D:\\workspace\\flink-demo\\flink-scala\\src\\main\\resources\\sensor.txt")

    // Source3: kafka
    val properties = new Properties();
    properties.setProperty("bootstrap.servers", "e71:9092")
    properties.setProperty("group.id", "Acc1")
//    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
//    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
//    properties.setProperty("auto.offset.reset", "latest")
    properties.setProperty("auto.offset.reset", "earliest")

    val stream3 = env.addSource(new FlinkKafkaConsumer011[String]("sensor2", new SimpleStringSchema(), properties))

    // 3. Transformation

    // 4. Sink
//    stream1.print("stream1").setParallelism(1);
    stream3.print("stream3").setParallelism(1);

    env.execute("API test");
  }

}

